/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package com.bieber.node;

import com.bieber.common.*;
import com.bieber.common.model.ConsumerPerson;
import com.bieber.common.node.IgniteServer;
import com.bieber.node.pojo.Organization;
import com.bieber.node.pojo.Person;
import org.apache.ignite.*;
import org.apache.ignite.cache.affinity.*;
import org.apache.ignite.cache.query.*;
import org.apache.ignite.lang.*;

import javax.cache.*;
import java.util.*;

/**
 * Cache queries example. This example demonstrates SQL, TEXT, and FULL SCAN
 * queries over cache.
 * <p>
 * Example also demonstrates usage of fields queries that return only required
 * fields instead of whole key-value pairs. When fields queries are distributed
 * across several nodes, they may not work as expected. Keep in mind following
 * limitations (not applied if data is queried from one node only):
 * <ul>
 *     <li>
 *         Joins will work correctly only if joined objects are stored in
 *         collocated mode. Refer to {@link org.apache.ignite.cache.affinity.AffinityKey} javadoc for more details.
 *     </li>
 *     <li>
 *         Note that if you created query on to replicated cache, all data will
 *         be queried only on one node, not depending on what caches participate in
 *         the query (some data from partitioned cache can be lost). And visa versa,
 *         if you created it on partitioned cache, data from replicated caches
 *         will be duplicated.
 *     </li>
 * </ul>
 * <p>
 * Remote nodes should be started using {@link ExampleNodeStartup} which will
 * start node with {@code examples/config/example-ignite.xml} configuration.
 */
public class CacheQueryExample {
    /** Organizations cache name. */
    private static final String ORG_CACHE = CacheQueryExample.class.getSimpleName() + "Organizations";

    /** Persons cache name. */
    private static final String PERSON_CACHE = CacheQueryExample.class.getSimpleName() + "Persons";

    /**
     * Executes example.
     *
     * @param args Command line arguments, none required.
     * @throws Exception If example execution failed.
     */
    public static void main(String[] args) throws Exception {
        Ignition.setClientMode(true);
        Ignite ignite = Ignition.start(IgniteServer.class.getResource("/ignite-server.xml"));
        System.out.println();
        System.out.println(">>> Cache query example started.");

        sqlQuery();
        print("Cache query example finished.");
    }

    /**
     * Example for scan query based on a predicate.
     */
    private static void scanQuery() {
        IgniteCache<Integer, ConsumerPerson> cache = Ignition.ignite().cache(Constants.PERSON_CACHE);

        ScanQuery<Integer, ConsumerPerson> scan = new ScanQuery<Integer, ConsumerPerson>(
                new IgniteBiPredicate<Integer, ConsumerPerson>() {
                    @Override public boolean apply(Integer key, ConsumerPerson person) {
                        return true;
                    }
                }
        );

        // Execute queries for salary ranges.
        print("People with salaries between 0 and 1000 (queried with SCAN query): ", cache.query(scan).getAll());
    }

    /**
     * Example for SQL queries based on salary ranges.
     */
    private static void sqlQuery() {
        IgniteCache<Integer, PersonOrder> cache = Ignition.ignite().cache(Constants.PERSON_ORDER_CACHE);

        // SQL clause which selects salaries based on range.
        String sql = "1=1 order by amount desc";

        // Execute queries for salary ranges.
        print("People with salaries between 0 and 1000 (queried with SQL query): ",
                cache.query(new SqlQuery<Integer, PersonOrder>(PersonOrder.class, sql)).getAll());
    }

    /**
     * Example for SQL queries based on all employees working for a specific organization.
     */
    private static void sqlQueryWithJoin() {
        IgniteCache<AffinityKey<UUID>, Person> cache = Ignition.ignite().cache(PERSON_CACHE);

        // SQL clause query which joins on 2 types to select people for a specific organization.
        String joinSql =
                "from Person, \"" + ORG_CACHE + "\".Organization as org " +
                        "where Person.orgId = org.id " +
                        "and lower(org.name) = lower(?)";

        // Execute queries for find employees for different organizations.
        print("Following people are 'ApacheIgnite' employees: ",
                cache.query(new SqlQuery<AffinityKey<UUID>, Person>(Person.class, joinSql).
                        setArgs("ApacheIgnite")).getAll());

        print("Following people are 'Other' employees: ",
                cache.query(new SqlQuery<AffinityKey<UUID>, Person>(Person.class, joinSql).
                        setArgs("Other")).getAll());
    }

    /**
     * Example for TEXT queries using LUCENE-based indexing of people's resumes.
     */
    private static void textQuery() {
        IgniteCache<AffinityKey<UUID>, Person> cache = Ignition.ignite().cache(PERSON_CACHE);

        //  Query for all people with "Master Degree" in their resumes.
        QueryCursor<Cache.Entry<AffinityKey<UUID>, Person>> masters =
                cache.query(new TextQuery<AffinityKey<UUID>, Person>(Person.class, "Master"));

        // Query for all people with "Bachelor Degree" in their resumes.
        QueryCursor<Cache.Entry<AffinityKey<UUID>, Person>> bachelors =
                cache.query(new TextQuery<AffinityKey<UUID>, Person>(Person.class, "Bachelor"));

        print("Following people have 'Master Degree' in their resumes: ", masters.getAll());
        print("Following people have 'Bachelor Degree' in their resumes: ", bachelors.getAll());
    }

    /**
     * Example for SQL queries to calculate average salary for a specific organization.
     */
    private static void sqlQueryWithAggregation() {
        IgniteCache<AffinityKey<UUID>, Person> cache = Ignition.ignite().cache(PERSON_CACHE);

        // Calculate average of salary of all persons in ApacheIgnite.
        // Note that we also join on Organization cache as well.
        String sql =
                "select avg(salary) " +
                        "from Person, \"" + ORG_CACHE + "\".Organization as org " +
                        "where Person.orgId = org.id " +
                        "and lower(org.name) = lower(?)";

        QueryCursor<List<?>> cursor = cache.query(new SqlFieldsQuery(sql).setArgs("ApacheIgnite"));

        // Calculate average salary for a specific organization.
        print("Average salary for 'ApacheIgnite' employees: ", cursor.getAll());
    }

    /**
     * Example for SQL-based fields queries that return only required
     * fields instead of whole key-value pairs.
     */
    private static void sqlFieldsQuery() {
        IgniteCache<AffinityKey<UUID>, Person> cache = Ignition.ignite().cache(PERSON_CACHE);

        // Execute query to get names of all employees.
        QueryCursor<List<?>> cursor = cache.query(new SqlFieldsQuery(
                "select concat(firstName, ' ', lastName) from Person"));

        // In this particular case each row will have one element with full name of an employees.
        List<List<?>> res = cursor.getAll();

        // Print names.
        print("Names of all employees:", res);
    }

    /**
     * Example for SQL-based fields queries that return only required
     * fields instead of whole key-value pairs.
     */
    private static void sqlFieldsQueryWithJoin() {
        IgniteCache<AffinityKey<UUID>, Person> cache = Ignition.ignite().cache(PERSON_CACHE);

        // Execute query to get names of all employees.
        String sql =
                "select concat(firstName, ' ', lastName), org.name " +
                        "from Person, \"" + ORG_CACHE + "\".Organization as org " +
                        "where Person.orgId = org.id";

        QueryCursor<List<?>> cursor = cache.query(new SqlFieldsQuery(sql));

        // In this particular case each row will have one element with full name of an employees.
        List<List<?>> res = cursor.getAll();

        // Print persons' names and organizations' names.
        print("Names of all employees and organizations they belong to:", res);
    }

    /**
     * Populate cache with test data.
     */
    private static void initialize() {
        IgniteCache<Integer, Organization> orgCache = Ignition.ignite().cache(ORG_CACHE);

        // Organizations.
        Organization org1 = new Organization("ApacheIgnite");
        Organization org2 = new Organization("Other");

        orgCache.put(org1.getId(), org1);
        orgCache.put(org2.getId(), org2);

        IgniteCache<Integer, Person> personCache = Ignition.ignite().cache(PERSON_CACHE);

        // People.
        Person p1 = new Person(org1, "John", "Doe", 2000, "John Doe has Master Degree.");
        Person p2 = new Person(org1, "Jane", "Doe", 1000, "Jane Doe has Bachelor Degree.");
        Person p3 = new Person(org2, "John", "Smith", 1000, "John Smith has Bachelor Degree.");
        Person p4 = new Person(org2, "Jane", "Smith", 2000, "Jane Smith has Master Degree.");

        // Note that in this example we use custom affinity key for Person objects
        // to ensure that all persons are collocated with their organizations.
        personCache.put(p1.getId(), p1);
        personCache.put(p2.getId(), p2);
        personCache.put(p3.getId(), p3);
        personCache.put(p4.getId(), p4);
    }

    /**
     * Prints message and query results.
     *
     * @param msg Message to print before all objects are printed.
     * @param col Query results.
     */
    private static void print(String msg, Iterable<?> col) {
        print(msg);
        print(col);
    }

    /**
     * Prints message.
     *
     * @param msg Message to print before all objects are printed.
     */
    private static void print(String msg) {
        System.out.println();
        System.out.println(">>> " + msg);
    }

    /**
     * Prints query results.
     *
     * @param col Query results.
     */
    private static void print(Iterable<?> col) {
        for (Object next : col)
            System.out.println(">>>     " + next);
    }

}
